package com.mjf.transformation

import org.apache.flink.api.common.RuntimeExecutionMode
import org.apache.flink.streaming.api.functions.co.CoFlatMapFunction
import org.apache.flink.streaming.api.scala._
import org.apache.flink.util.Collector

object CoFlatMapDemo {
  def main(args: Array[String]): Unit = {

    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
    env.setParallelism(1)
    env.setRuntimeMode(RuntimeExecutionMode.BATCH)

    val source1: DataStream[(String, Int)] = env.fromCollection(List(("tina", 23),("paul", 30)))

    val source2: DataStream[(String, String)] = env.fromCollection(List(("lucky", "female"),("paul", "male")))

    source1.print("source1")
    source2.print("source2")

    val connectStream: ConnectedStreams[(String, Int), (String, String)] = source1.connect(source2)

    val result: DataStream[String] = connectStream.flatMap(new MyCoFlatMapFunction)

    result.print("result")

    env.execute(CoFlatMapDemo.getClass.getName)

  }
}

class MyCoFlatMapFunction extends CoFlatMapFunction[(String, Int), (String, String), String] {
  override def flatMap1(value: (String, Int), out: Collector[String]): Unit = out.collect(value._1 + "的年龄：" + value._2)

  override def flatMap2(value: (String, String), out: Collector[String]): Unit = out.collect(value._1 + "的性别：" + value._2)
}
